package com.htdata.iiot.capture;
import java.io.UnsupportedEncodingException;
import java.math.BigInteger;
import java.net.URI;
import java.net.URISyntaxException;
import java.text.SimpleDateFormat;
import java.util.Properties;
import java.util.Timer;
import java.util.TimerTask;
import org.apache.log4j.Logger;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.MqttSecurityException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.json.JSONArray;
import org.json.JSONObject;
public class Bridge{
	private Logger logger;
	private static BridgeProfile bridgeProfile = com.htdata.iiot.capture.BridgeProfile.getBridgeProfile();
	public  MqttConnectOptions mosquitoConnectOptions  = null;
	public  MqttConnectOptions emqttConnectOptions     = null;
	private MqttClient  mosquitoClient;
	private MqttClient  emqttClient;
	private MqttMessage message;
	private MqttTopic   topic;
	private Long startTime = 0L;
	private Long msgCount = 0L;
	private Long msgAnchor = 0L;
	private String 		emqBrokerList[];
	private String 		emqClientId;
	private String 		emqttPassWord;
	private String 		emqttUsrName;
	private String 		emqttTopic;
	private int 		emqttQos;
	
	private int[] 		mosquitoQos;
	private String[] 	mosBrokerList;
	private String 		mosClientId;
	private String[] 	mosquitoBroker;
	private String 		mosquitoUserName;
	private String 		mosquitoPassWord;
	private String[] 	mosquitoTopic;
	private String[] 	mosquitoKeyArray;
	
	private boolean     flag = true;
	
	
	
	public Bridge() {
		logger = Logger.getLogger(Bridge.class);
		mosBrokerList 	 = bridgeProfile.getMosquitoBroker();
		mosClientId 	 = bridgeProfile.getMosquitoClientId();
		emqBrokerList	 = bridgeProfile.getEmqttBroker();
		emqClientId  	 = bridgeProfile.getEmqttClientId();
		emqttPassWord    = bridgeProfile.getEmqttPassWord();
		emqttUsrName     = bridgeProfile.getEmqttUserName();
		mosquitoBroker   = bridgeProfile.getMosquitoBroker();
		mosquitoUserName = bridgeProfile.getMosquitoUserName();
		mosquitoPassWord = bridgeProfile.getMosquitoPassWord();
		mosquitoQos      = bridgeProfile.getMosquitoQos();
		mosquitoTopic    = bridgeProfile.getMosquitoTopic();
		emqttTopic       = bridgeProfile.getEmqttTopic();
		emqttQos         = bridgeProfile.getEmqttQos();
		mosquitoKeyArray = bridgeProfile.getMosquitoKeyArray();
	}
	
	private void prepareMqtt() {
		
		
		while (true) {
			try {
				if(mosquitoClient == null)
				      mosquitoClient = new MqttClient(mosBrokerList[0], mosClientId);
				if (mosquitoClient != null) {
					logger.debug("connected to mosquito successfully");
					break;
				}
				logger.warn("create emqtt MqttClient failed!");
			} catch (MqttException e) {
				logger.warn("create mosquito MqttClient failed! " + e.getMessage());
			}
			try {
				Thread.sleep(200);
			} catch (InterruptedException e) {				
			}
		}
		
		while (true) {
			try {
				if(emqttClient == null)
					emqttClient = new MqttClient(emqBrokerList[0], emqClientId);
				
				if (emqttClient != null) {
					logger.info("connected to emqtt successfully");
					break;
				}
				logger.warn("create emqtt MqttClient failed!");
			} catch (MqttException e) {
				logger.warn("create emqtt MqttClient failed! " + e.getMessage());
			}
			try {
				Thread.sleep(200);
			} catch (InterruptedException e) {
				
			}
		}
	}
	
	private void startConnect() {
		emqttConnectOptions = new MqttConnectOptions();
		emqttConnectOptions.setAutomaticReconnect(true);
		emqttConnectOptions.setCleanSession(true);
		emqttConnectOptions.setUserName(emqttUsrName);
		emqttConnectOptions.setPassword(emqttPassWord.toCharArray());
			
	    mosquitoConnectOptions = new MqttConnectOptions();
		mosquitoConnectOptions.setServerURIs(mosquitoBroker);
		mosquitoConnectOptions.setAutomaticReconnect(true);
		mosquitoConnectOptions.setCleanSession(false);
		mosquitoConnectOptions.setMaxInflight(100);
		mosquitoConnectOptions.setUserName(mosquitoUserName);
		mosquitoConnectOptions.setPassword(mosquitoPassWord.toCharArray());
		
		
		//mosquitoClient 回调函数
		try {
			mosquitoClient.setCallback(new MqttCallback() {
				
				public void connectionLost(Throwable cause) {
					
	                 while (!mosquitoClient.isConnected()) {
	                     try {
	                         Thread.sleep(1000);
	                         mosquitoClient.connect(mosquitoConnectOptions);;
	                     } catch (Exception e) {
	                    	 logger.error("mosquitoClient again connect false:"+e.getMessage());
	                     }
	                 }
				}
				public void deliveryComplete(IMqttDeliveryToken token) {
					
				}
				public void messageArrived(String string, MqttMessage msg)
						throws Exception {
					mosqMessgArrived(string,msg);
				}
			});
			
			mosquitoClient.connect(mosquitoConnectOptions);
			mosquitoClient.subscribe(mosquitoTopic, mosquitoQos);
			
		} catch (Exception e) {
			logger.warn("mosquitoClient:"+e.getMessage());
		}
		
		//emqttClient 回调函数
		try {
			emqttClient.setCallback(new MqttCallback() {
				
			    public void connectionLost(Throwable cause) {
				
				     logger.error(cause.getMessage());
					 while (!emqttClient.isConnected()) {
					     try {
					         Thread.sleep(1000);
					         emqttClient.connect(emqttConnectOptions);
					     } catch (Exception e) {
					    	 logger.error("emqttClient again connect false:"+e.getMessage());
					     }
					 }
				}
			    
				public void deliveryComplete(IMqttDeliveryToken token) {
		
				}
				
				public void messageArrived(String string, MqttMessage msg)
						throws Exception {
				}
			});
					
			emqttClient.connect(emqttConnectOptions);
			
		} catch (Exception e) {
			logger.warn("emqttClient"+e.getMessage());
		}
		
		
		if (bridgeProfile.getBenchmark()) {
			
			startTime = System.currentTimeMillis();
			
			TimerTask timerTask = new TimerTask() {
				@Override
				public void run() {
					long curTime = System.currentTimeMillis();
					long count = msgCount.longValue();
					long dec = count - msgAnchor;
					long avg = (count * 1000) / (curTime - startTime);
					msgAnchor = count;
					logger.info(String.format("mqtt:{last:%8d, total:%8d, avg:%8d, tid:%d}", dec, count, avg,
							Thread.currentThread().getId()));
				}
			};
			Timer timer = new Timer();
			timer.scheduleAtFixedRate(timerTask, 1000, 1000);
		}
	}
	public void Run() {
		
		prepareMqtt();
		startConnect();
		
		while(true){
			if(flag == false)
				break;
		}
		try {
			emqttClient.close();
		} catch (MqttException e) {
		}
		
		try {
			mosquitoClient.close();
		} catch (MqttException e) {
		}
	}
	
	synchronized private void IncrementCount() {
		++msgCount;
	}
	
	 public  String date2TimeStamp(String date,String format){  
         try {  
             SimpleDateFormat sdf = new SimpleDateFormat(format);  
             return  String.valueOf(sdf.parse(date).getTime());      
         } catch (Exception e) {  
              e.printStackTrace();  
          }  
         return "";  
     }  
	
	
	public void mosqMessgArrived(String strTopic,MqttMessage message) {
		
		Long   tsTime = 0L;
		
		IncrementCount();
		
		byte[] payload = message.getPayload();
		if (payload.length == 0) {
			return;
		}
		strTopic = strTopic.substring(1);
		JSONObject jsonSub = new JSONObject(new String(payload));
		if(jsonSub.has("sensorDatas")){
			JSONArray jsonArray = jsonSub.getJSONArray("sensorDatas");
			
			if(mosquitoKeyArray.length < (jsonArray.length()-1)){
				logger.error("key length < jsonArray length,please check mosqutio2emqtt.properties");
				return ;
			}
			
			StringBuffer strBuf = new StringBuffer();			
			tsTime = System.currentTimeMillis();
			strBuf.append("{\"dn\":\""+strTopic+"\",\"ts\":"+tsTime);
			
			JSONObject obj =jsonArray.getJSONObject(0);																			
			String valueStrSrc = obj.getString("value");
			double valueSrc = Double.parseDouble(valueStrSrc);
			double valueDst = 125*valueSrc/16 - 56.25;
			String valueStrDst = String.valueOf(valueDst);
			strBuf.append(",\""+mosquitoKeyArray[0]+"\":"+valueStrDst);

/*			
			for(int i = 0 ;i<jsonArray.length()-1;i++){
					JSONObject obj =jsonArray.getJSONObject(i);																			
					String valueStr = obj.getString("value");
					strBuf.append(",\""+mosquitoKeyArray[i]+"\":"+valueStr);					
			}  
*/
			strBuf.append("}");
			pub2Emqtt(strBuf.toString());
			   
		} 
	}
	
	public  void pub2Emqtt(String data) {
		
		this.message = new MqttMessage();
		this.message.setQos(emqttQos);
		this.message.setRetained(true);
		this.message.setPayload(data.getBytes());
		this.topic = emqttClient.getTopic(emqttTopic);
		MqttDeliveryToken token;
		
		try {
			token = this.topic.publish(this.message);
			token.waitForCompletion();
		} catch (MqttPersistenceException e) {
			logger.warn(e.getMessage());
		} catch (MqttException e) {
			logger.warn(e.getMessage());
		}
	}
		
}